package models

import (
	"encoding/json"
	"fmt"
	"ginclassmates/utils"
	"log"

	"github.com/Shopify/sarama"
)

// 不指定消费组，指定某一个分区进行消费
func KafkaComsumerinit() error {
	config := sarama.NewConfig()
	config.Version = sarama.V3_3_1_0
	config.Consumer.Offsets.Initial = sarama.OffsetNewest // 从最新的 offset 开始读取
	consumer, err := sarama.NewConsumer([]string{utils.Kafkaaddr}, config)
	if err != nil {
		// 错误处理
		//log.Fatalln(err)
		return err
	}
	//defer consumer.Close()

	partition := utils.Kafkapartition // 指定分区号为 2
	offset := sarama.OffsetNewest     // 从最早的 offset 开始读取
	partitionConsumer, err := consumer.ConsumePartition(utils.Kafkatopic, partition, offset)
	if err != nil {
		// 错误处理
		fmt.Println(err)
	}
	//defer partitionConsumer.Close()
	go func() {
		for msg := range partitionConsumer.Messages() {
			log.Printf("分区是: %d, 偏移量是: %d, key: %s, Value: %s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
			signin_User := Signin_User{}
			//将获取的值反编译成结构体数据
			if err := json.Unmarshal(msg.Value, &signin_User); err != nil {
				log.Println("数据解析失败！" + err.Error())
			} else {
				//把新纪录添加到数据库里面
				err = Deposit(signin_User.Name, signin_User.Number, signin_User.Class)
				if err != nil {
					log.Println("数据添加失败！" + err.Error())
				}
				//获取以往的考勤记录
				data, err := Query_Signin(signin_User.Number)
				if err != nil {
					log.Println(err)
					//将处理的结果发回到订阅频道中
					err = RedisPublish(signin_User.Number, err)
					if err != nil {
						log.Println("返回数据发送失败！" + err.Error())
					}
				} else {
					msg, err := json.Marshal(data)
					if err != nil {
						log.Printf("返回数据发送失败！" + err.Error())
						err = RedisPublish(signin_User.Number, err)
						if err != nil {
							log.Println("打包考勤记录发送失败！" + err.Error())
						}
					}
					//将处理的结果发回到订阅频道中
					err = RedisPublish(signin_User.Number, string(msg))
					if err != nil {
						log.Printf("返回数据发送失败！" + err.Error())
					}
				}

			}
		}
	}()

	return nil
}
